package com.chief.state;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class StateBackends {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//        env.setStateBackend(new EmbeddedRocksDBStateBackend());

        env.setStateBackend(new HashMapStateBackend());

//        env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("viewfs://mycluster/data1/flink-cp/"));
//        env.getCheckpointConfig().setCheckpointStorage("viewfs://mycluster/data1/aaa/");

//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop52:8020/flink-cp2/");
        env.getCheckpointConfig().setCheckpointStorage("file:///E:/2222/");
//        env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());

        env.getCheckpointConfig().setCheckpointInterval(3000);

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(5)));


        env.setParallelism(1);

        DataStreamSource<String> streamSource = env.socketTextStream("hadoop55", 7777);

        SingleOutputStreamOperator<Integer> map = streamSource.map(new OperatorStateTest.MyOperatorMapFunction(33.33));

        map.print();

        env.execute("StateBackends");

    }
}
